/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.service.client.impl;

import java.util.Hashtable;

import cgl.narada.matching.Profile;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.client.NBEventListener;
import cgl.narada.service.client.NBNotificationListener;
import cgl.narada.service.client.NBRecoveryListener;
import cgl.narada.service.client.NBRecoveryNotification;
import cgl.narada.service.qos.ConsumerConstraints;
import cgl.narada.service.qos.impl.ConsumerConstraintsImpl;
import cgl.narada.service.qos.impl.QosServiceImpl;
import cgl.narada.service.replay.EntityReplayActuator;
import cgl.narada.service.replay.ReplayRequest;
import cgl.narada.service.replay.ReplayServiceListener;

/**
 * The event consumer provides a host of functions for clients to consume
 * events. These include subscription to profiles and also the specification of
 * QoS constraints that an event must satisfy prior to delivery. The consumer
 * also includes several utlity functions among them are the ability to retrieve
 * the number of events received at this consumer.
 * 
 * @author Shrideep Pallickara $Date$ $Revision$
 * 
 */

public class EventConsumerImpl implements EventConsumer {

    private int numOfEventsReceivedSoFar = 0;

    private long timeOfLastEvent = 0;

    /** Keeps track of specified qos constraints */
    private Hashtable constraints;

    /** Keeps track of all the profiles */
    private Hashtable profiles;

    /** The event listener for this consumer */
    private NBEventListener nbEventListener;

    /** The event notification listener for this consumer */
    private NBNotificationListener nbNotificationListener;

    private int consumerId = 0;

    private QosServiceImpl qosServiceImpl;

    private String moduleName = "EventConsumerImpl: ";

    private int entityId = 0;

    private Hashtable recoveryListeners;

    protected EventConsumerImpl(int entityId, int consumerId,
            QosServiceImpl qosServiceImpl) {
        this.entityId = entityId;
        this.consumerId = consumerId;
        this.qosServiceImpl = qosServiceImpl;
        constraints = new Hashtable();
        profiles = new Hashtable();
        recoveryListeners = new Hashtable();
    }

    public int getEntityId() {
        return entityId;
    }

    /** Retrieves the number of events routed by this event consumer */
    public int getNumberOfEventsReceivedSoFar() {
        return numOfEventsReceivedSoFar;
    }

    /**
     * Retrieves the time elapsed (in milliseconds) since the last event was
     * received on this consumer
     */
    public long getTimeElapsedSinceLastEvent() {
        return (System.currentTimeMillis() - timeOfLastEvent);
    }

    /** Checks to see if a constraint has been registered for a given profile */
    public boolean hasConsumerConstraints(Profile profile)
            throws ServiceException {
        String key = profile.getProfileId();
        if (constraints.containsKey(key)) {
            Object obj = constraints.get(key);
            if (obj == null) {
                return false;
            }
            return true;
        }
        return false;
    }

    /**
     * Verify if the specify QosConstraints can indeed be satisfied. This method
     * will true if there aare no specified Qos associated with the profile.
     */
    public boolean canSatisfyQosRequirements(Profile profile)
            throws ServiceException {
        /** NBNI */
        System.out.println(moduleName
                + "The canSatisfyQosRequirements(profile) "
                + "needs to be implemented ");
        return false;
    }

    /**
     * Retrieves the QosEnablerAndSpecifer for a given profile. Returns null if
     * none have been specified for the profile in question or if the profile
     * has not yet been registered to this consumer.
     */
    public ConsumerConstraints getConsumerConstraints(Profile profile)
            throws ServiceException {
        if (hasConsumerConstraints(profile)) {
            String key = profile.getProfileId();
            ConsumerConstraints qosSpecifier = (ConsumerConstraints) constraints
                    .get(key);
            return qosSpecifier;
        }

        return null;
    }

    /** Gets a list of profiles managed by the consumer */
    public Profile[] getListOfManagedProfiles() {
        return (Profile[]) profiles.values().toArray(new Profile[0]);
    }

    /** Sets the event listener for the profiles managed by this consumer */
    public void setNBEventListener(NBEventListener nbEventListener)
            throws ServiceException {
        this.nbEventListener = nbEventListener;
    }

    /** Gets the event listener for the profiles managed by this consumer */
    public NBEventListener getNBEventListener() {
        return nbEventListener;
    }

    /**
     * Sets the notification listener for the profiles (and the specified Qos
     * constraints) managed by this consumer
     */
    public void setNBNotificationListener(
            NBNotificationListener nbNotificationListener)
            throws ServiceException {
        this.nbNotificationListener = nbNotificationListener;
    }

    /** Create consumer constraints associated with a certain profile */
    public ConsumerConstraints createConsumerConstraints(Profile profile)
            throws ServiceException {
        if (profile == null) {
            throw new ServiceException(moduleName + "Specified profile is NULL");
        }

        ConsumerConstraintsImpl consumerConstraints = new ConsumerConstraintsImpl(
                entityId, consumerId, profile, qosServiceImpl
                        .generateConstraintId());

        return consumerConstraints;
    }

    /**
     * Subscribe to a profile and register a listener to receive events
     * conforming to the specified constraint
     */
    public void subscribeTo(Profile profile) throws ServiceException {
        if (profile == null) {
            throw new ServiceException(moduleName + "Specified profile is NULL");
        }
        String key = profile.getProfileId();

        if (key == null) {
            throw new ServiceException(moduleName + "Profile ID is NULL for"
                    + profile);
        }

        if (profiles.containsKey(key)) {
            throw new ServiceException(moduleName
                    + "Profile is already registered" + profile);
        }

        profiles.put(key, profile);
        qosServiceImpl.subscribeTo(profile, null, consumerId);
    }

    /**
     * Unsubscribes a profile. If there is a registered ConsumerConstraints for
     * this profile, it is also disabled.
     */
    public void unSubscribe(Profile profile) throws ServiceException {
        if (profile == null) {
            throw new ServiceException(moduleName + "Specified profile is NULL");
        }
        String key = profile.getProfileId();

        if (key == null) {
            throw new ServiceException(moduleName + "Profile ID is NULL for"
                    + profile);
        }

        if (!profiles.containsKey(key)) {
            throw new ServiceException(moduleName + "Profile [" + profile
                    + "] not registered previously" + " faulty Unsubscription");
        }

        profiles.remove(key);
        constraints.remove(key);

        qosServiceImpl.unsubscribe(profile);
    }

    /**
     * Subscribe to a profile, but request that the event satisfies some QoS
     * constraints prior to delivery. This subscription may trigger a discovery
     * request and based upon the responses or lack thereof the subscription
     * propagation may succeed or not. This method requires that a
     * NotificationListener be registered prior to activating the profile.
     */
    public void subscribeTo(Profile profile,
            ConsumerConstraints consumerConstraints) throws ServiceException {

        if (profile == null || consumerConstraints == null) {
            throw new ServiceException(moduleName + "Specified profile = "
                    + profile + "qosEnablerAndSpecifier = "
                    + consumerConstraints);
        }
        String key = profile.getProfileId();

        if (key == null) {
            throw new ServiceException(moduleName + "Profile ID is NULL for"
                    + profile);
        }

        if (profiles.containsKey(key)) {
            throw new ServiceException(moduleName
                    + "Profile is already registered " + profile);
        }

        String _key = consumerConstraints.getProfile().getProfileId();

        if (!_key.equals(key)) {
            throw new ServiceException(moduleName + "Constraints have been "
                    + "specified on a profile different from the "
                    + "one you are trying to subscribe to");
        }

        profiles.put(key, profile);
        constraints.put(key, consumerConstraints);
        qosServiceImpl.subscribeTo(profile, consumerConstraints, consumerId);

    }

    /**
     * This is called by an entity to recover from failures or prolonged
     * disconnects
     */
    public long recover(int templateId, NBRecoveryListener recoveryListener)
            throws ServiceException {
        if (recoveryListener == null) {
            throw new ServiceException(moduleName
                    + "Specified recoveryListener is NULL");
        }

        if (templateId == 0) {
            throw new ServiceException(moduleName
                    + "Wrong templateId=0 specified for recovery");
        }

        long recoveryId = qosServiceImpl.recoverEntity(consumerId, entityId,
                templateId, false);
        Object recoveryKey = new Long(recoveryId);
        recoveryListeners.put(recoveryKey, recoveryListener);

        return recoveryId;
    }

    public void propagateRecoveryNotification(
            NBRecoveryNotification recoveryNotification) {
        long recoveryId = recoveryNotification.getRecoveryId();
        Object recoveryKey = new Long(recoveryId);

        NBRecoveryListener recoveryListener = (NBRecoveryListener) recoveryListeners
                .get(recoveryKey);

        if (recoveryListener == null) {
            System.out.println(moduleName + "RecoveryListener associated with "
                    + recoveryNotification + " is NULL");
            return;
        }

        recoveryListener.onRecovery(recoveryNotification);
    }

    /** Initiate replay based on the created replay request */
    public void initiateReplay(ReplayRequest replayRequest,
            ReplayServiceListener replayListener) throws ServiceException {
        EntityReplayActuator entityReplayActuator = cgl.narada.service.replay.impl.EntityReplayActuatorImpl
                .getInstance();

        entityReplayActuator.initiateReplay(replayRequest, replayListener);
    }

}
